package com.bfxy.kafka.api.consumer.commit;

import com.bfxy.kafka.api.constant.Const;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

public class CommitConsumer {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.218.21:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "commit-group");
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
//        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
//        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

        // 使用手动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
            consumer.subscribe(Collections.singletonList(Const.TOPIC_COMMIT));
            System.out.println("commit consumer started...");
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (TopicPartition partition : records.partitions()) {
                    List<ConsumerRecord<String, String>> partitionRecordList = records.records(partition);
                    String topic = partition.topic();
                    int size = partitionRecordList.size();
                    System.out.println(String.format("--- 获取topic：%s，分区位置：%s，消息总数：%s ---", topic, partition.partition(), size));
                    for (ConsumerRecord<String, String> consumerRecord : partitionRecordList) {
                        String value = consumerRecord.value();
                        long offset = consumerRecord.offset();
                        long commitOffset = offset + 1;
                        System.out.println(String.format("-- 获取实际消息value：%s，消息offset：%s，提交offset：%s ---", value, offset, commitOffset));
                        // 实际工作中都是在一个partition内部，每一条消息记录进行一一提交
                        // 实际提交的offset肯定是比实际的消息的offset大1，因为实际提交的offset是下次要拉取的消息的offset
                        // 单条消息记录提交：同步方式
//                        consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(commitOffset)));
                        // 单条消息记录提交：异步方式
                        consumer.commitAsync(Collections.singletonMap(partition, new OffsetAndMetadata(commitOffset)), new OffsetCommitCallback() {
                            @Override
                            public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                                if (null != e) {
                                    System.out.println("异步提交报错...");
                                }
                                System.out.println("异步提交成功：" + map);
                            }
                        });
                    }
                    /*// 一个partition做一次提交动作
                    long latestOffsetInPartition = partitionRecordList.get(partitionRecordList.size() - 1).offset();
                    long commitLatestOffsetInPartition = latestOffsetInPartition + 1;
                    // 分区提交：同步方式
//                    consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(commitLatestOffsetInPartition)));
                    // 分区提交：异步方式
                    consumer.commitAsync(Collections.singletonMap(partition, new OffsetAndMetadata(commitLatestOffsetInPartition)), new OffsetCommitCallback() {
                        @Override
                        public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                            if (null != e) {
                                System.out.println("异步提交报错...");
                            }
                            System.out.println("异步提交成功：" + map);
                        }
                    });*/
                }
                /*// 整体提交：同步方式
//                consumer.commitSync();
                // 整体提交：异步方式
//                consumer.commitAsync();
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                        if (null != e) {
                            System.out.println("异步提交报错...");
                        }
                        System.out.println("异步提交成功：" + map);
                    }
                });*/

            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
